package com.rickie.pvuv.streaming;

import com.rickie.pvuv.config.Constants;
import com.rickie.pvuv.domain.UserBehaviorEvent;
import com.rickie.pvuv.domain.UserBehaviorEventSchema;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

import java.util.Properties;

/**
 * 处理用户访问记录 pvuv
 */
public class UserBehaviorProcessJob {
    public static void main(String[] args) throws Exception {
        //获取Flink 的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //Checkpoint 配置，默认情况下，state会保存在taskmanager的内存中，checkpoint会存储在JobManager的内存中。
        //每隔5000ms启动一个检查点（设置CheckPoint的周期）
        env.enableCheckpointing(5000);
        //设置模式为.EXACTLY_ONCE（默认值），还可以设置为AT_LEAST_ONCE
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //checkpoint的最小间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        //检查点必须在1min内完成，或者被丢弃（checkPoint的超时时间）
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        //同一时间只允许执行一个检查点
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        //表示Flink处理程序被cancel后，会保留CheckPoint数据，以便根据实际需要恢复到指定的CheckPoint
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        // Kafka参数
        String topic = Constants.TOPIC;
        Properties prop = new Properties();
        prop.setProperty("bootstrap.servers",Constants.KAFKA_BOOTSTRAP_SERVER);
        prop.setProperty("group.id","test.group");

        FlinkKafkaConsumer<UserBehaviorEvent> myConsumer = new FlinkKafkaConsumer<>(topic, new UserBehaviorEventSchema(), prop);
        //默认消费策略
        myConsumer.setStartFromGroupOffsets();
        //设置Kafka数据源
        DataStreamSource<UserBehaviorEvent> userBehaviorStream = env.addSource(myConsumer);
        // 打印输出
        userBehaviorStream.print().setParallelism(1);
        // execute
        env.execute("UserBehaviorProcessJob");
    }
}
